package com.birdy.rocketmq;

import com.birdy.dto.Response;
import com.birdy.mq.FacekillGroupMqFactory;
import com.birdy.mq.MqProperties;
import com.birdy.mq.con.MqConst;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.LitePullConsumer;
import org.apache.rocketmq.client.consumer.MQConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;

/**
 * @author birdy
 * @date 2022/4/5 10:18 PM
 */
@Slf4j
@RestController
@RequestMapping("/rocket/")
public class RocketMqAction {
    @Autowired
    private MqProperties mqProperties;
    @Autowired
    private FacekillGroupMqFactory factory;
    @GetMapping("send")
    public Object sendMessage(@RequestParam String message) {
        try {
            log.info("{}, {}", mqProperties.getGroup(), mqProperties.getNameSrvAddr());
            MQProducer mqProducer = factory.getProducer();
            Message msg = new Message(MqConst.TOPIC_COMMON,
                    message.getBytes(RemotingHelper.DEFAULT_CHARSET)); // 消息需要二进制
            SendResult send = mqProducer.send(msg);
//            mqProducer.shutdown();
            return send;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
    @GetMapping("pull_get")
    public Object pullGetMessage() {
        Response.ResponseBuilder builder = Response.builder();
        try {
            List<MessageExt> messages = new ArrayList<>();
            LitePullConsumer consumer = factory.getLitePullConsumer();
            consumer.subscribe(MqConst.TOPIC_COMMON, "*");
            consumer.start(); // 启动消息的消费者
            boolean hasMessage = true;
            while (hasMessage) {
                List<MessageExt> messageList = consumer.poll();
                System.out.println(messageList);
                if (messageList != null && messageList.size() != 0) {
                    messages = messageList;
                } else {
                    hasMessage = false;
                }
            }
//            consumer.shutdown();
            return builder.res(Response.SUCCESS).msg("success").data("msglist", messages).build();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return builder.res(Response.FAILED).msg("failed").build();

    }
    @GetMapping("get")
    public Object getMessage() {
        Response.ResponseBuilder builder = Response.builder();
        try {
            List<String> messages = new ArrayList<>();
            DefaultMQPushConsumer consumer = factory.getPushConsumer();
            consumer.subscribe(MqConst.TOPIC_COMMON, "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() { // 准备接收消息
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    msgs.forEach((msg)->{
                        messages.add(new String(msg.getBody()));
                        System.out.printf("【%s】接收到新的消息：body = %s、level = %s %n", Thread.currentThread().getName(), new String(msg.getBody()), msg.getProperty("level"));
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 消息消费成功
                }
            });
            consumer.start(); // 启动消息的消费者
            return builder.res(Response.SUCCESS).msg("success").data("msglist", messages).build();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return builder.res(Response.FAILED).msg("failed").build();
    }

    @GetMapping("hello")
    public Object hello() {
        return "hello";
    }
}
